package com.xzc.apitest.udftest

import com.xzc.apitest.source.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala._
import org.apache.flink.table.functions.AggregateFunction
import org.apache.flink.types.Row

object AggregateFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //1.11之后默认就是这个，之前 OlePlanner
    val settings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    //创建表环境
    val tableEnv = StreamTableEnvironment.create(env, settings)

    val inputStream = env.readTextFile("D:\\git\\learning_flink\\_01_试用\\src\\main\\resources\\sensor.txt")

    //将流转换为样例类形式
    val dataStream = inputStream
      .map(data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }).assignTimestampsAndWatermarks(
      //在转换为表之前就将流的watermark指定，从而在表里面可以直接使用
      new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(3)) {
        override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L
      }
    )

    //建立动态表，最后的字段是flink给的时间特性
    //    val sensorTable = tableEnv.fromDataStream(dataStream,
    //      'id, 'temperature, 'timestamp, 'pt.proctime)
    //这里的rowtime就是上面流里面定义的watermark，覆盖了原有的timestamp
    val sensorTable = tableEnv.fromDataStream(dataStream,
      'id, 'temperature, 'timestamp.rowtime as 'ts)

    //table api
    val avgTempFun = new AvgTempFun
    val resultTable = sensorTable
      .groupBy('id)
      .aggregate(avgTempFun('temperature) as 'avgTemp)
      .select('id, 'avgTemp)

    //sql
    tableEnv.createTemporaryView("sensor", sensorTable)
    tableEnv.registerFunction("avgTempFun", avgTempFun)
    val resultSqlTable = tableEnv.sqlQuery(
      """
        |select id,avgTempFun(temperature)
        |from
        |sensor
        |group by id
        |""".stripMargin
    )

    resultTable.toRetractStream[Row].print("result")
    resultSqlTable.toRetractStream[Row].print("sql")

    env.execute("aggregate function test")
  }
}

//定义一个类，专门用于表示聚合的状态
//因为元组是无法更新其值的，所以这里要对象来代替，每次来新数据就更新
class AvgTempAcc {
  var sum: Double = 0.0
  var count: Int = 0
}

//自定义一个聚合函数，求每个传感器的平均温度值，保存状态(tempSum, tempCount)，直接是结果，后面累加计算就可以了
class AvgTempFun extends AggregateFunction[Double, AvgTempAcc] {
  //计算平均值，用总量/总个数
  override def getValue(accum: AvgTempAcc): Double = accum.sum / accum.count

  //初始值
  override def createAccumulator(): AvgTempAcc = new AvgTempAcc

  //还要实现一个具体的处理accum的方法
  def accumulate(accum: AvgTempAcc, temp: Double): Unit = {
    accum.sum += temp
    accum.count += 1
  }
}
